/**
 * Copyright (c) 2010 Yahoo! Inc. All rights reserved. 
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you
 * may not use this file except in compliance with the License. You
 * may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0 
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied. See the License for the specific language governing
 * permissions and limitations under the License. See accompanying
 * LICENSE file.
 */

package workloads;

import generator.CounterGenerator;
import generator.DiscreteGenerator;
import generator.Generator;
import generator.IntegerGenerator;
import generator.ScrambledZipfianGenerator;
import generator.SkewedLatestGenerator;
import generator.UniformIntegerGenerator;
import generator.ZipfianGenerator;

import java.util.Properties;

import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.Random;
import java.util.Vector;
import java.util.concurrent.ConcurrentHashMap;

import measurements.Measurements;

import tlsb.*;
import util.Configuration;
import util.TextFileLoader;

/**
 * The core benchmark scenario. Represents a set of clients doing simple CRUD
 * operations. The relative proportion of different kinds of operations, and
 * other properties of the workload, are controlled by parameters specified at
 * runtime.
 * 
 * Properties to control the client:
 * <UL>
 * <LI><b>fieldcount</b>: the number of fields in a record (default: 10)
 * <LI><b>fieldlength</b>: the size of each field (default: 100)
 * <LI><b>readallfields</b>: should reads read all fields (true) or just one
 * (false) (default: true)
 * <LI><b>writeallfields</b>: should updates and read/modify/writes update all
 * fields (true) or just one (false) (default: false)
 * <LI><b>readproportion</b>: what proportion of operations should be reads
 * (default: 0.95)
 * <LI><b>updateproportion</b>: what proportion of operations should be updates
 * (default: 0.05)
 * <LI><b>insertproportion</b>: what proportion of operations should be inserts
 * (default: 0)
 * <LI><b>scanproportion</b>: what proportion of operations should be scans
 * (default: 0)
 * <LI><b>readmodifywriteproportion</b>: what proportion of operations should be
 * read a record, modify it, write it back (default: 0)
 * <LI><b>requestdistribution</b>: what distribution should be used to select
 * the records to operate on - uniform, zipfian or latest (default: uniform)
 * <LI><b>maxscanlength</b>: for scans, what is the maximum number of records to
 * scan (default: 1000)
 * <LI><b>scanlengthdistribution</b>: for scans, what distribution should be
 * used to choose the number of records to scan, for each scan, between 1 and
 * maxscanlength (default: uniform)
 * <LI><b>insertorder</b>: should records be inserted in order by key
 * ("ordered"), or in hashed order ("hashed") (default: hashed)
 * </ul>
 */
public class CoreWorkload extends Workload {

	/**
	 * The name of the database table to run queries against.
	 */
	public static final String TABLENAME_PROPERTY = "table";

	/**
	 * The default name of the database table to run queries against.
	 */
	public static final String TABLENAME_PROPERTY_DEFAULT = "usertable";

	public static String table;
	public static String uid = "";
	public static String mid = "";
	public static String fid = "";
	public static int start = 0;

	/**
	 * The name of the property for the number of fields in a record.
	 */
	public static final String FIELD_COUNT_PROPERTY = "fieldcount";

	/**
	 * Default number of fields in a record.
	 */
	public static final String FIELD_COUNT_PROPERTY_DEFAULT = "10";

	int fieldcount;

	/**
	 * The name of the property for the length of a field in bytes.
	 */
	public static final String FIELD_LENGTH_PROPERTY = "fieldlength";

	/**
	 * The default length of a field in bytes.
	 */
	public static final String FIELD_LENGTH_PROPERTY_DEFAULT = "100";

	int fieldlength;

	/**
	 * The name of the property for deciding whether to read one field (false)
	 * or all fields (true) of a record.
	 */
	public static final String READ_ALL_FIELDS_PROPERTY = "readallfields";

	/**
	 * The default value for the readallfields property.
	 */
	public static final String READ_ALL_FIELDS_PROPERTY_DEFAULT = "true";

	boolean readallfields;

	/**
	 * The name of the property for deciding whether to write one field (false)
	 * or all fields (true) of a record.
	 */
	public static final String WRITE_ALL_FIELDS_PROPERTY = "writeallfields";

	/**
	 * The default value for the writeallfields property.
	 */
	public static final String WRITE_ALL_FIELDS_PROPERTY_DEFAULT = "false";

	boolean writeallfields;

	Random random = new Random();
	/*
	 * ConcurrentHashMap<Integer, String> uidchm = Utils
	 * .getConcurrentHashMap(Configuration.instance().getReadSequenceFile());
	 */
	TextFileLoader uidchm = new TextFileLoader(Configuration.instance()
			.getReadSequenceFile(), true, Configuration.instance()
			.getQueueThreshold());

	/**
	 * The name of the property for the the distribution of requests across the
	 * keyspace. Options are "uniform", "zipfian" and "latest"
	 */
	public static final String REQUEST_DISTRIBUTION_PROPERTY = "requestdistribution";

	/**
	 * The default distribution of requests across the keyspace
	 */
	public static final String REQUEST_DISTRIBUTION_PROPERTY_DEFAULT = "uniform";

	public static final String POST_PROPORTION_PROPERTY = "postproportion";
	public static final String POST_PROPORTION_PROPERTY_DEFAULT = "0.0";
	public static final String FOLLOW_PROPORTION_PROPERTY = "followproportion";
	public static final String FOLLOW_PROPORTION_PROPERTY_DEFAULT = "0.0";
	public static final String UNFOLLOW_PROPORTION_PROPERTY = "unfollowproportion";
	public static final String UNFOLLOW_PROPORTION_PROPERTY_DEFAULT = "0.0";
	public static final String NEWUSER_PROPORTION_PROPERTY = "newuserproportion";
	public static final String NEWUSER_PROPORTION_PROPERTY_DEFAULT = "0.0";
	// public static final String PUSH_PROPORTION_PROPERTY="pushproportion";
	// public static final String PUSH_PROPORTION_PROPERTY_DEFAULT="0.0";
	// public static final String PULL_PROPORTION_PROPERTY="pullproportion";
	// public static final String PULL_PROPORTION_PROPERTY_DEFAULT="0.0";
	// public static final String HYBRID_PROPORTION_PROPERTY="bybridproportion";
	// public static final String HYBRID_PROPORTION_PROPERTY_DEFAULT="0.0";
	public static final String HTL_PROPORTION_PROPERTY = "htlproportion";
	public static final String HTL_PROPORTION_PROPERTY_DEFAULT = "0.0";

	IntegerGenerator keysequence;

	DiscreteGenerator operationchooser;

	IntegerGenerator keychooser;

	Generator fieldchooser;

	CounterGenerator transactioninsertkeysequence;

	IntegerGenerator scanlength;

	boolean orderedinserts;

	int recordcount;

	/**
	 * Initialize the scenario. Called once, in the main client thread, before
	 * any operations are started.
	 */
	public void init(Properties p) throws WorkloadException {
		uidchm.start();
		table = p.getProperty(TABLENAME_PROPERTY, TABLENAME_PROPERTY_DEFAULT);
		fieldcount = Integer.parseInt(p.getProperty(FIELD_COUNT_PROPERTY,
				FIELD_COUNT_PROPERTY_DEFAULT));
		fieldlength = Integer.parseInt(p.getProperty(FIELD_LENGTH_PROPERTY,
				FIELD_LENGTH_PROPERTY_DEFAULT));
		double postproportion = Double.parseDouble(p.getProperty(
				POST_PROPORTION_PROPERTY, POST_PROPORTION_PROPERTY_DEFAULT));
		double followproportion = Double
				.parseDouble(p.getProperty(FOLLOW_PROPORTION_PROPERTY,
						FOLLOW_PROPORTION_PROPERTY_DEFAULT));
		double unfollowproportion = Double.parseDouble(p.getProperty(
				UNFOLLOW_PROPORTION_PROPERTY,
				UNFOLLOW_PROPORTION_PROPERTY_DEFAULT));
		double newuserproportion = Double.parseDouble(p.getProperty(
				NEWUSER_PROPORTION_PROPERTY,
				NEWUSER_PROPORTION_PROPERTY_DEFAULT));
		// double
		// pushproportion=Double.parseDouble(p.getProperty(PUSH_PROPORTION_PROPERTY,PUSH_PROPORTION_PROPERTY_DEFAULT));
		// double
		// pullproportion=Double.parseDouble(p.getProperty(PULL_PROPORTION_PROPERTY,PULL_PROPORTION_PROPERTY_DEFAULT));
		// double
		// hybridproportion=Double.parseDouble(p.getProperty(HYBRID_PROPORTION_PROPERTY,HYBRID_PROPORTION_PROPERTY_DEFAULT));
		double htlproportion = Double.parseDouble(p.getProperty(
				HTL_PROPORTION_PROPERTY, HTL_PROPORTION_PROPERTY_DEFAULT));
		recordcount = Integer.parseInt(p
				.getProperty(Client.RECORD_COUNT_PROPERTY));
		String requestdistrib = p.getProperty(REQUEST_DISTRIBUTION_PROPERTY,
				REQUEST_DISTRIBUTION_PROPERTY_DEFAULT);

		int insertstart = Integer.parseInt(p.getProperty(INSERT_START_PROPERTY,
				INSERT_START_PROPERTY_DEFAULT));

		readallfields = Boolean.parseBoolean(p.getProperty(
				READ_ALL_FIELDS_PROPERTY, READ_ALL_FIELDS_PROPERTY_DEFAULT));
		writeallfields = Boolean.parseBoolean(p.getProperty(
				WRITE_ALL_FIELDS_PROPERTY, WRITE_ALL_FIELDS_PROPERTY_DEFAULT));

		keysequence = new CounterGenerator(insertstart);
		operationchooser = new DiscreteGenerator();
		if (postproportion > 0) {
			operationchooser.addValue(postproportion, "POST");
		}

		if (followproportion > 0) {
			operationchooser.addValue(followproportion, "FOLLOW");
		}

		if (unfollowproportion > 0) {
			operationchooser.addValue(unfollowproportion, "UNFOLLOW");
		}

		if (newuserproportion > 0) {
			operationchooser.addValue(newuserproportion, "NEWUSER");
		}

		// if (pushproportion>0)
		// {
		// operationchooser.addValue(pushproportion,"PUSH");
		// }
		//
		// if (pullproportion>0)
		// {
		// operationchooser.addValue(pullproportion,"PULL");
		// }
		//
		// if (hybridproportion>0)
		// {
		// operationchooser.addValue(hybridproportion,"HYBRID");
		// }

		if (htlproportion > 0) {
			operationchooser.addValue(htlproportion, "HTL");
		}

		transactioninsertkeysequence = new CounterGenerator(recordcount);
		if (requestdistrib.compareTo("uniform") == 0) {
			keychooser = new UniformIntegerGenerator(0, recordcount - 1);
		} else if (requestdistrib.compareTo("zipfian") == 0) {
			// it does this by generating a random "next key" in part by taking
			// the modulus over the number of keys
			// if the number of keys changes, this would shift the modulus, and
			// we don't want that to change which keys are popular
			// so we'll actually construct the scrambled zipfian generator with
			// a keyspace that is larger than exists at the beginning
			// of the test. that is, we'll predict the number of inserts, and
			// tell the scrambled zipfian generator the number of existing keys
			// plus the number of predicted keys as the total keyspace. then, if
			// the generator picks a key that hasn't been inserted yet, will
			// just ignore it and pick another key. this way, the size of the
			// keyspace doesn't change from the perspective of the scrambled
			// zipfian generator

			int opcount = Integer.parseInt(p
					.getProperty(Client.OPERATION_COUNT_PROPERTY));
			int expectednewkeys = (int) (((double) opcount) * postproportion * 2.0); // 2
																						// is
																						// fudge
																						// factor

			keychooser = new ScrambledZipfianGenerator(recordcount
					+ expectednewkeys);
		} else if (requestdistrib.compareTo("latest") == 0) {
			keychooser = new SkewedLatestGenerator(transactioninsertkeysequence);
		} else {
			throw new WorkloadException("Unknown distribution \""
					+ requestdistrib + "\"");
		}

		fieldchooser = new UniformIntegerGenerator(0, fieldcount - 1);
	}

	/**
	 * Do one insert operation. Because it will be called concurrently from
	 * multiple client threads, this function must be thread safe. However,
	 * avoid synchronized, or the threads will block waiting for each other, and
	 * it will be difficult to reach the target throughput. Ideally, this
	 * function would have no side effects other than DB operations.
	 */
	public boolean doPost(DB db, Object threadstate) {
		int keynum = keysequence.nextInt();
		if (!orderedinserts) {
			keynum = Utils.hash(keynum);
		}
		String dbkey = "user" + keynum;
		HashMap<String, String> values = new HashMap<String, String>();
		for (int i = 0; i < fieldcount; i++) {
			String fieldkey = "field" + i;
			String data = Utils.ASCIIString(fieldlength);
			values.put(fieldkey, data);
		}
		if (db.post(uid, mid) == 0)
			return true;
		else
			return false;
	}

	/**
	 * Do one transaction operation. Because it will be called concurrently from
	 * multiple client threads, this function must be thread safe. However,
	 * avoid synchronized, or the threads will block waiting for each other, and
	 * it will be difficult to reach the target throughput. Ideally, this
	 * function would have no side effects other than DB operations.
	 */
	public boolean doTransaction(DB db, Object threadstate) {
		String op = operationchooser.nextString();

		if (op.compareTo("FOLLOW") == 0) {
			doTransactionFollow(db);
		} else if (op.compareTo("UNFOLLOW") == 0) {
			doTransactionUnfollow(db);
		} else if (op.compareTo("NEWUSER") == 0) {
			doTransactionNewUser(db);
		} else if (op.compareTo("POST") == 0) {
			doTransactionPost(db);
		}
		// else if (op.compareTo("PUSH")==0)
		// {
		// doTransactionPush(db);
		// }
		// else if (op.compareTo("PULL")==0)
		// {
		// doTransactionPull(db);
		// }
		// else if (op.compareTo("HYBRID")==0)
		// {
		// doTransactionHybrid(db);
		// }
		else if (op.compareTo("HTL") == 0) {
			doTransactionGetHTL(db);
		}

		return true;
	}

	public void doTransactionFollow(DB db) {

		db.follow(uid, fid);
	}

	public void doTransactionUnfollow(DB db) {

		db.unfollow(uid, fid);
	}

	public void doTransactionNewUser(DB db) {

		db.newUser();
	}

	public void doTransactionPost(DB db) {
		// if( Utils.getCurList().get(0).size()==0){
		// Utils.getCurList().pop();
		// }
		String status = Utils.getCurList();
		if (status != null) {
			String[] tmp = status.split("\t");
			db.post(tmp[1], tmp[0]);
		} else {
			System.err.println("Utils.getCurList return null");
		}
	}

	// public void doTransactionPush(DB db)
	// {
	//
	// db.getByPush(uid);
	// }
	// public void doTransactionPull(DB db)
	// {
	//
	// db.getByPull(uid);
	// }
	// public void doTransactionHybrid(DB db)
	// {
	//
	// db.getByHybrid(uid);
	// }
	public void doTransactionGetHTL(DB db) {
		// uid = uidchm.get(random.nextInt(uidchm.size()));
		uid = uidchm.readLine();
		db.getHTL(uid, start);
	}

}
